package com.jsy.mqhelper.charlotterpc.core.mq.rabbit;

import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetAddress;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.BlockedCallback;
import com.rabbitmq.client.BlockedListener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ExceptionHandler;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.UnblockedCallback;
import com.rabbitmq.client.impl.AMQConnection;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.ObjectPool;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;

import static com.jsy.mqhelper.common.constant.Constants.MQ_HELPER;

/**
 * @author: SongyangJi
 * @description:
 * @since: 2021/9/27
 */
@Slf4j(topic = MQ_HELPER)
public class PooledChannelConnectionFactory extends RabbitConnectionFactory {

    com.rabbitmq.client.Connection connection;

    public PooledChannelConnectionFactory(ConnectionFactory connectionFactory) {
        super(connectionFactory);
    }

    public synchronized Connection createConnection() throws IOException, TimeoutException {
        if (this.connection == null || !this.connection.isOpen()) {
            connection = new ConnectionWrapper(bareConnectionFactory.newConnection());
        }
        // TODO connection.addShutdownListener
        connection.addShutdownListener(new ShutdownListener() {
            @Override
            public void shutdownCompleted(ShutdownSignalException cause) {
                log.info("ShutdownSignalException cause={}", cause.getReason());
            }
        });

        if (connection instanceof AMQConnection) {
            log.info("local port={}", ((AMQConnection)connection).getLocalPort());
        }
        return connection;
    }

    private static class ConnectionWrapper implements com.rabbitmq.client.Connection {

        private final ObjectPool<Channel> channels;
        Connection delegate;

        public ConnectionWrapper(Connection delegate) {
            this.delegate = delegate;
            this.channels = new GenericObjectPool<>(new ChannelFactory());
        }

        @Override
        public void addShutdownListener(ShutdownListener listener) {
            delegate.addShutdownListener(listener);
        }

        @Override
        public void removeShutdownListener(ShutdownListener listener) {
            delegate.removeShutdownListener(listener);
        }

        @Override
        public ShutdownSignalException getCloseReason() {
            return delegate.getCloseReason();
        }

        @Override
        public void notifyListeners() {
            delegate.notifyListeners();
        }

        @Override
        public boolean isOpen() {
            return delegate.isOpen();
        }

        @Override
        public InetAddress getAddress() {
            return delegate.getAddress();
        }

        @Override
        public int getPort() {
            return delegate.getPort();
        }

        @Override
        public int getChannelMax() {
            return delegate.getChannelMax();
        }

        @Override
        public int getFrameMax() {
            return delegate.getFrameMax();
        }

        @Override
        public int getHeartbeat() {
            return delegate.getHeartbeat();
        }

        @Override
        public Map<String, Object> getClientProperties() {
            return delegate.getClientProperties();
        }

        @Override
        public String getClientProvidedName() {
            return delegate.getClientProvidedName();
        }

        @Override
        public Map<String, Object> getServerProperties() {
            return delegate.getServerProperties();
        }

        // todo
        @SneakyThrows
        @Override
        public Channel createChannel() throws IOException {
            log.debug("borrow channel from pool");
            log.debug("idle objects num={}", this.channels.getNumIdle());
            log.debug("active objects num={}", this.channels.getNumActive());
            return this.channels.borrowObject();
        }

        @Override
        public Channel createChannel(int channelNumber) throws IOException {
            return delegate.createChannel(channelNumber);
        }

        @Override
        public void close() throws IOException {
            delegate.close();
        }

        @Override
        public void close(int closeCode, String closeMessage) throws IOException {
            delegate.close(closeCode, closeMessage);
        }

        @Override
        public void close(int timeout) throws IOException {
            delegate.close(timeout);
        }

        @Override
        public void close(int closeCode, String closeMessage, int timeout) throws IOException {
            delegate.close(closeCode, closeMessage, timeout);
        }

        @Override
        public void abort() {
            delegate.abort();
        }

        @Override
        public void abort(int closeCode, String closeMessage) {
            delegate.abort(closeCode, closeMessage);
        }

        @Override
        public void abort(int timeout) {
            delegate.abort(timeout);
        }

        @Override
        public void abort(int closeCode, String closeMessage, int timeout) {
            delegate.abort(closeCode, closeMessage, timeout);
        }

        @Override
        public void addBlockedListener(BlockedListener listener) {
            delegate.addBlockedListener(listener);
        }

        @Override
        public BlockedListener addBlockedListener(BlockedCallback blockedCallback,
            UnblockedCallback unblockedCallback) {
            return delegate.addBlockedListener(blockedCallback, unblockedCallback);
        }

        @Override
        public boolean removeBlockedListener(BlockedListener listener) {
            return delegate.removeBlockedListener(listener);
        }

        @Override
        public void clearBlockedListeners() {
            delegate.clearBlockedListeners();
        }

        @Override
        public ExceptionHandler getExceptionHandler() {
            return delegate.getExceptionHandler();
        }

        @Override
        public String getId() {
            return delegate.getId();
        }

        @Override
        public void setId(String id) {
            delegate.setId(id);
        }

        private Channel bareCreateChannel() throws IOException {
            return delegate.createChannel();
        }

        private Channel createChannelProxy(Channel channel) {
            return (Channel)Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {Channel.class},
                new ChannelProxyHandler(channel));
        }

        private class ChannelFactory implements PooledObjectFactory<Channel> {

            @Override
            public PooledObject<Channel> makeObject() throws Exception {
                Channel channel = createChannelProxy(bareCreateChannel());
                return new DefaultPooledObject<>(channel);
            }

            @Override
            public void destroyObject(PooledObject<Channel> p) throws Exception {
                p.getObject().close();
            }

            @Override
            public boolean validateObject(PooledObject<Channel> p) {
                return p.getObject().isOpen();
            }

            @Override
            public void activateObject(PooledObject<Channel> p) throws Exception {

            }

            @Override
            public void passivateObject(PooledObject<Channel> p) throws Exception {

            }
        }

        private class ChannelProxyHandler implements InvocationHandler {
            Channel channel;

            public ChannelProxyHandler(Channel channel) {
                this.channel = channel;
            }

            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                if (method.getName().equals("close")) {
                    log.debug("return channel to pool");
                    ConnectionWrapper.this.channels.returnObject((Channel)proxy);
                    return null;
                }
                return method.invoke(channel, args);
            }
        }

    }

}
